package sos.entrega2013.mr1.noviembre.bo.fraude;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import sos.entrega2013.mr1.noviembre.beans.CDR;

public class FraudeReduce extends MapReduceBase implements Reducer<Text, CDR, Text, Text> {
	//----
	private final double  	VELOCIDAD_MAXIMA=1000;
	private boolean 		bIsFraude=false;
	private final int MILESEGUNDOS_HORA=3600000;
	private double dTiempo1;
	private double dTiempo2;
	private double dDistancia;
	private double velocidad;
	private int    iX1=0;
	private int    iX2=0;
	private int    iY1=0;
	private int    iY2=0;
	@Override
	public void reduce(Text key, Iterator<CDR> values,OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
		
		
		CDR usuario			= values.next();		
		//numero de telefono del usuario
		String sNumUsuario	= key.toString();
		while(values.hasNext()){
			CDR llamada=values.next();
			//obtenemos las coordenadas de la primera posicion 
			if (usuario.getLlamante().equals(sNumUsuario)){
				iX1=usuario.getCoordXllamante();
				iY1=usuario.getCoordYllamante();				
			}
			else if(usuario.getLlamado().equals(sNumUsuario)){
				iX1=usuario.getCoordXllamado();
				iY1=usuario.getCoordYllamado();
			}
			//obtnemos las coordenadas de la segunda posicion
			if (llamada.getLlamante().equals(sNumUsuario)){
				iX2=llamada.getCoordXllamante();
				iY2=llamada.getCoordYllamante();				
			}
			else if(llamada.getLlamado().equals(sNumUsuario)){
				iX2=llamada.getCoordXllamado();
				iY2=llamada.getCoordYllamado();
			}
			//---------------------------------------------------------------------------
			this.dDistancia	=	Math.sqrt(Math.pow((iX2-iX1),2) - Math.pow((iY2-iY1),2));
			this.dTiempo1	=	usuario.getFecha().getTime();
			this.dTiempo2	=	llamada.getFecha().getTime();
			
			this.velocidad	=	this.dDistancia/((this.dTiempo2-this.dTiempo1)/this.MILESEGUNDOS_HORA);
			if(this.velocidad > this.VELOCIDAD_MAXIMA){
				this.bIsFraude	=	true;
			}
			usuario=llamada;
		}
		
		if(this.bIsFraude){
			output.collect(key, new Text("alarma-fraude"));
		}
	}

}
